package com.sduept.nwld.dataserver.manager.message;

import com.sduept.component.message.core.UserAlarmMessageService;
import com.sduept.component.message.model.TopicSubscriptionInfo;
import com.sduept.core.dao.SuperDAO;
import com.sduept.generator.ObjectId;
import com.sduept.kafka.KafkaTopicConfig;
import com.sduept.kafka.model.KafkaTopic;
import com.sduept.message.entity.Message;
import com.sduept.nwld.dataview.controller.message.UserSubscribedTopicDto;
import com.sduept.permission.entity.PersonnelEntityBean;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.persistence.TypedQuery;
import java.util.*;

@Service
public class UserMessageManager {

    @Autowired
    private SuperDAO dao;

    @Autowired
    private UserAlarmMessageService service;
    @PersistenceContext(unitName = "primaryUnit")
    protected EntityManager em;
    @Autowired
    private KafkaTopicConfig kafkaTopicConfig;

    /**
     * 设置当前当前类型的消息已阅读
     *
     * @param type 消息类型
     * @param id   消息id
     */
    public void markMessageReaded(String type, String id) {
        try {
            Message am = (Message) dao.findById(id, Class.forName("com.sduept.component.message.model." + type));
            am.setStatus(1);
            dao.createOrUpdate(am, Message.class);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    /**
     * 按条件查询消息
     *
     * @param esql
     * @param cla
     * @param orderby
     * @param range
     * @return
     */
    public List<Message> findMessageRecordByCondition(String esql, Class<? extends Message> cla, String orderby, int[] range) {
        String sql = "FROM  " + cla.getSimpleName();
        if (!StringUtils.isEmpty(esql)) {
            sql = sql + " WHERE " + esql;
        }
        if (!StringUtils.isEmpty(orderby)) {
            sql = sql + orderby;
        }
        @SuppressWarnings("unchecked")
        TypedQuery<Message> query = (TypedQuery<Message>) em.createQuery(sql, cla);
        if (range != null && range[1] > range[0]) {
            query.setMaxResults(range[1] - range[0]);
            query.setFirstResult(range[0]);
        }
        return query.getResultList();
    }

    /**
     * 计算消息的记录数用于分页
     *
     * @param esql
     * @param cla
     * @return
     */
    public int countMessageRecords(String esql, Class<? extends Message> cla) {
        String sql = "SELECT COUNT(*) FROM  " + cla.getSimpleName();
        if (!StringUtils.isEmpty(esql)) {
            sql = sql + " WHERE " + esql;
        }
        TypedQuery<Long> query = em.createQuery(sql, Long.class);
        return query.getSingleResult().intValue();
    }

    /**
     * 获得当前用户的订阅信息
     *
     * @param user
     * @return
     */
    public List<UserSubscribedTopicDto> getUserSubscribedTopicDtos(PersonnelEntityBean user) {
        List<UserSubscribedTopicDto> result = new ArrayList<UserSubscribedTopicDto>();
        List<KafkaTopic> messageTopicList = getKafkaTopicListByScopes(new String[]{"user"});
        for (KafkaTopic topic : messageTopicList) {
            UserSubscribedTopicDto dto = new UserSubscribedTopicDto();
            dto.setName(topic.getName());
            dto.setType(topic.getTopicType());
            dto.setDescription(topic.getDescription());
            if (checkUserHadSubscriptThisTopic(topic.getName(), user.getCode()))
                dto.setCheck(true);
            else
                dto.setCheck(false);
            result.add(dto);
        }
        return result;
    }

    /**
     * 保存用户订阅信息
     *
     * @param dto
     * @param user
     */
    public void saveUserSubscriptionInfo(UserSubscribedTopicDto dto, PersonnelEntityBean user) {
        if (dto.isCheck()) {
            TopicSubscriptionInfo info = service.getTopicInfoMap().get(dto.getName());
            if (info != null) {
                if (!info.getSubscriber().contains(user.getCode())) {
                    info.setSubscriber(info.getSubscriber() + UserAlarmMessageService.split_remark + user.getCode());
                    service.updateTopicSubscriptionInfo(info);
                }
            } else {
                info = new TopicSubscriptionInfo();
                info.setId(ObjectId.get().toString());
                info.setSubscriber(user.getCode());
                info.setTopic(dto.getName());
                service.updateTopicSubscriptionInfo(info);
            }
        } else {
            TopicSubscriptionInfo info = service.getTopicInfoMap().get(dto.getName());
            if (info != null) {
                if (info.getSubscriber().contains(user.getCode())) {
                    String newSubscriber = "";
                    if (info.getSubscriber().indexOf(UserAlarmMessageService.split_remark + user.getCode()) != -1) {
                        newSubscriber = info.getSubscriber().replace(UserAlarmMessageService.split_remark + user.getCode(), "");
                    } else if (info.getSubscriber().indexOf(user.getCode() + UserAlarmMessageService.split_remark) != -1) {
                        newSubscriber = info.getSubscriber().replace(user.getCode() + UserAlarmMessageService.split_remark, "");
                    } else if (info.getSubscriber().indexOf(user.getCode()) != -1) {
                        newSubscriber = info.getSubscriber().replace(user.getCode(), "");
                    } else {
                        return;
                    }
                    info.setSubscriber(newSubscriber);
                    service.updateTopicSubscriptionInfo(info);
                }
            }
        }
    }

    /**
     * 设置当前用户当前类型的消息全部已阅读
     *
     * @param type 消息类型
     * @param uid  用户id
     */
    public void markCurrentTypeMessageAllReaded(String type, String uid) {
        try {
            @SuppressWarnings("unchecked")
            List<Message> ms = (List<Message>) dao.findByCondition(Class.forName("com.sduept.message.entity." + type), "target=? and status=0", uid);
            for (Message m : ms) {
                m.setStatus(1);
                dao.createOrUpdate(m, Message.class);
            }
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获得指定范围的主题
     *
     * @param type
     * @return
     */
    public List<KafkaTopic> getKafkaTopicListByScopes(String[] scopes) {
        List<KafkaTopic> list = new ArrayList<KafkaTopic>();
        Map<String, KafkaTopic> col = kafkaTopicConfig.getTopics();
//        Collection<KafkaTopic> col = KafkaProxy.getInstace().getLegitimateTopics();
        for (KafkaTopic t : col.values()) {
            for (String scope : scopes) {
                if (!StringUtils.isEmpty(scope) && t.getScope().contains(scope))
                    list.add(t);
            }
        }
        return list;
    }

    /**
     * 获得用户未读的消息
     *
     * @param cla
     * @param uid
     * @return
     */
    public List<Message> getUnReadMessage(Class<? extends Message> cla, String uid) {
        List<Message> result = new ArrayList<Message>();
        List<? extends Message> list = dao.findByCondition(cla, "target=?1 and status=0", uid);
        for (Message m : list) {
            if (m.getCreateTime().getTime() + m.getTimeliness() > (new Date().getTime()))
                result.add(m);
        }
        return result;
    }

    /**
     * 检验当前用户是否订阅了当前主题的消息
     *
     * @param topic
     * @param uid
     * @return
     */
    private boolean checkUserHadSubscriptThisTopic(String topic, String uid) {
        TopicSubscriptionInfo info = service.getTopicInfoMap().get(topic);
        if (info != null && info.getSubscriber().contains(uid))
            return true;
        return false;
    }
}
